package com.smsc.headend.task.engine.handler.impl.metercontrol;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.map.MapUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.metercontrol.enums.ControlState;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.protocol.ansi.AnsiData;
import com.smsc.headend.module.protocol.cosem.bo.CosemData;
import com.smsc.headend.module.report.entity.LogConnectDisConnectHis;
import com.smsc.headend.module.report.entity.LogModeSwitchHisSub;
import com.smsc.headend.module.report.entity.LogScheduleTaskDevice;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.SubTaskGroup;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.mapper.LogConnectDisConnectHisMapper;
import com.smsc.headend.task.engine.service.CosemXmlService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.task.LogScheduleTaskDeviceService;
import com.smsc.headend.task.engine.utils.AnsiUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;

/**
 * 目前检查拉合闸状态 及 拉闸子步骤第一步检查状态都进入该 handler
 */
@Slf4j
public class ReadControlStateHandler implements TaskHandler {
    @Autowired
    CosemXmlService cosemXmlService;
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    LogConnectDisConnectHisMapper logConnectDisConnectHisMapper;
    @Autowired
    LogScheduleTaskDeviceService logScheduleTaskDeviceService;

    @Override
    public int handle(Task task, AtomicTask atomicTask, DeviceData deviceData, String actionResult) throws UDISTaskHandlerException {
        if (!(deviceData instanceof CosemData) && !(deviceData instanceof AnsiData)) {
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.DEVICE_DATA_TYPE_NOT_SUPPORT.toString());
        }

        Integer state;
        if (deviceData instanceof AnsiData) {
            state = AnsiUtil.parseConnectDisconnect(((AnsiData) deviceData).getAnsiByteBuffer());
        } else {
            CosemData cosemData = (CosemData) deviceData;
            String stateCode = cosemData.getValue();
            String type = cosemData.getType();
            state = Convert.toInt(cosemXmlService.getData(type, stateCode));
        }

        String desc = ControlState.forId(state).getDesc();
        log.debug("controlState" + state);
        Map<String, Object> parameterMap = getParameterMap(task, atomicTask);
        if (state == ControlState.Disconnected.getCode() && atomicTask.getGroupId().longValue() == OperationGroup.Disconnect.getId().longValue()) {
            if (task.getTaskType().equals(TaskType.SwitchMeterPayment.toString())) {
                replaceSwitchAtomicTask(task, atomicTask);
                Long taskId = MapUtil.getLong(parameterMap, "taskId");
                assetManagementService.updateStepStatus(taskId, task.getMeterId(), LogModeSwitchHisSub.Step.Disconnect, SubTaskStatus.Success.getCode(), null, SystemClock.now() / 1000, null);
                return 1;
            } else {
                removeSetAtomicTaskIfExists(task, atomicTask);
                updateLog(task.getMessageId(), task.getMeterId(), LogConnectDisConnectHis.Result.success.getCode(), "");
                if (task.getTaskType().equals(TaskType.MeterPlanCdc.toString())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.CollectedSuccess.getCode());
                }
                taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), "result", "The meter is Disconnected", 600L, true);
                return 1;
            }
        } else if (OperationGroup.CheckCdc.getId().equals(Convert.toInt(atomicTask.getGroupId()))) {
            // 单独的getCdcStatus流程这里才需要把状态同步。拉闸 合闸流程在remoteActionHandler最后处理
            sendControlStateToAdapter(task, atomicTask, state);
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), "result", desc, 600L, true);
            return 1;
        } else if (state == ControlState.Connected.getCode() && atomicTask.getGroupId().longValue() == OperationGroup.ConnectArmed.getId().longValue()) {
            // 已经是合闸状态 后面不继续合闸 结束任务
            removeSetAtomicTaskIfExists(task, atomicTask);
            updateLog(task.getMessageId(), task.getMeterId(), LogConnectDisConnectHis.Result.success.getCode(), "");
            if (task.getTaskType().equals(TaskType.MeterPlanCdc.toString())) {
                logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.CollectedSuccess.getCode());
            }
            if (task.getTaskType().equals(TaskType.SwitchMeterPayment.toString())) {
                Long taskId = MapUtil.getLong(parameterMap, "taskId");
                assetManagementService.updateStepStatus(taskId, task.getMeterId(), LogModeSwitchHisSub.Step.Disconnect, SubTaskStatus.Success.getCode(), null, SystemClock.now() / 1000, null);
                return 1;
            } else {
                taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), "result", "The meter is Connected", 600L, true);
                return 1;
            }
        }

        return 1;
    }

    private void replaceSwitchAtomicTask(Task task, AtomicTask atomicTask) {
        redisUtils.del(RedisKeys.getProcessingTaskKey(task.getTaskNo()));
        List<SubTaskGroup> subTaskGroups = task.getSubTaskGroups();
        taskDAO.atomicTaskQueueAdd(atomicTask);
        subTaskGroups.stream().filter(subTaskGroup -> subTaskGroup.getGroupId() != OperationGroup.Disconnect.getId().longValue()).map(SubTaskGroup::getSubTasks).flatMap(x -> x.stream()).
                forEach(acTask -> {
                    taskDAO.atomicTaskQueueAdd(acTask);
                });
    }

    private void removeSetAtomicTaskIfExists(Task task, AtomicTask atomicTask) {
        if (atomicTask.getGroupId().longValue() == OperationGroup.Disconnect.getId()) {
            //pop disconnect remoteAction task
//            List<Object> taskList = redisUtils.lRange(RedisKeys.getProcessingTaskKey(task.getTaskNo()), 0, -1);
            AtomicTask disconnectTask = taskDAO.atomicTaskQueuePopReverse(task.getTaskNo());
            log.info("remove disconnect atomicTask {} ", disconnectTask.getAtomicTaskNo());
        } else if (atomicTask.getGroupId().longValue() == OperationGroup.ConnectArmed.getId()) {
            //pop ConnectArmed remoteAction task 结束合闸后续所有任务
            AtomicTask disconnectTask = taskDAO.atomicTaskQueuePopReverse(task.getTaskNo());
//            String termQueueId = taskDAO.getTerminalTaskListKey(task);
//            taskDAO.terminalTaskQueuePop(termQueueId);
            taskDAO.atomicTaskQueueDelete(task.getTaskNo());
            taskDAO.clearTaskProcessingAtomic(task.getTaskNo());
            log.info("remove harConnect atomicTask {} ", disconnectTask.getAtomicTaskNo());
        }
    }

    @Override
    public int handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error {}, ", message);
        if (atomicTask.getGroupId() == OperationGroup.ConnectArmed.getId().intValue()
                || atomicTask.getGroupId() == OperationGroup.Disconnect.getId().intValue()
                || atomicTask.getGroupId() == OperationGroup.CheckCdc.getId().intValue()) {
            if (task.getTaskType().equals(TaskType.SwitchMeterPayment.toString())) {
                Map<String, Object> parameterMap = getParameterMap(task, atomicTask);
                Long taskId = MapUtil.getLong(parameterMap, "taskId");
                assetManagementService.updateStepStatus(taskId, task.getMeterId(), atomicTask.getGroupId() == OperationGroup.ConnectArmed.getId().intValue() ? LogModeSwitchHisSub.Step.Connect : LogModeSwitchHisSub.Step.Disconnect,
                        SubTaskStatus.Failed.getCode(), null, SystemClock.now() / 1000, String.format("handle %s fail, %s", OperationName.forId(atomicTask.getOperationId()).getName(), message));
            } else {
                String key = "message";
                Map<String, Object> m = JSONUtil.toBean(task.getParamJson(), Map.class);
                key = m.get("seq") != null ? Convert.toStr(m.get("seq")) : key;
                String stepOperationName = OperationName.forId(atomicTask.getOperationId()).getName();
                if (task.getIsODR()) {
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), key, String.format("step %s failed, [%s]", stepOperationName, message), 3600l, false);
                }
                if (task.getTaskType().equals(TaskType.MeterPlanCdc.toString())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.DecodeDataFailed.getCode());
                }
                updateLog(task.getMessageId(), task.getMeterId(), LogConnectDisConnectHis.Result.failed.getCode(), message);
            }
        }
        return -1;
    }

    private void sendControlStateToAdapter(Task task, AtomicTask atomicTask, Integer state) {
        JSONObject obj = JSONUtil.createObj();
        if (atomicTask.getGroupId().longValue() == OperationGroup.CheckCdc.getId()) {
            obj.put("meterId", task.getMeterId());
            obj.put("value", Convert.toStr(state));
            obj.put("tv", SystemClock.now() / 1000);
            log.info("send control state to adapter: {}", JSONUtil.toJsonStr(obj));
            kafkaTemplate.send(TaskKafkaTopic.CLIENT_CONTROL_STATE, JSONUtil.toJsonStr(obj).getBytes(Charset.forName("utf-8")));
        }

    }

    public void updateLog(String messageId, Long meterId, Integer result, String reason) {
        List<LogConnectDisConnectHis> logConnectDisConnectHisList = logConnectDisConnectHisMapper.selectList(Wrappers.<LogConnectDisConnectHis>lambdaQuery()
                .eq(LogConnectDisConnectHis::getMessageId, messageId).eq(LogConnectDisConnectHis::getDeviceId, meterId).orderByDesc(LogConnectDisConnectHis::getTvUpdate));
        if (null != logConnectDisConnectHisList && !logConnectDisConnectHisList.isEmpty()) {
            LogConnectDisConnectHis logConnectDisConnectHis = logConnectDisConnectHisList.get(0);
            logConnectDisConnectHis.setResult(result);
            logConnectDisConnectHis.setReason(reason);
            logConnectDisConnectHisMapper.updateById(logConnectDisConnectHis);
        }
    }
}
